package ventpipe

import (
	"evtest/setting"
	"fmt"
	"time"
)

type PostMessage struct {
	ApiUrl     string
	JsonParam  []byte
	OutputChan chan *PipeRsp
}

type Ascend struct {
	pipe        *Pipe
	InputChan   chan *PostMessage
	ProcessChan chan *PostMessage
}

var AscendPipe *Ascend

func NewAscend(evConf *setting.EvConf) *Ascend {
	pipe := &Pipe{
		EvConf: evConf,
	}
	ascendPipe := &Ascend{
		pipe:        pipe,
		InputChan:   make(chan *PostMessage),
		ProcessChan: make(chan *PostMessage, 10),
	}
	return ascendPipe
}

func (asc *Ascend) RoutineRun() {
	go asc.filterTaskRun()
	go asc.pushTaskRun()
}

func (asc *Ascend) filterTaskRun() {
	for {
		select {
		case incomingMaterial := <-asc.InputChan:
			select {
			case asc.ProcessChan <- incomingMaterial:
				fmt.Printf("[%s] incoming material pop down: %v\n", time.Now().Format("2006-01-02 15:04:05"), incomingMaterial)
			default:
				fmt.Printf("[%s] incoming material drop: %v\n", time.Now().Format("2006-01-02 15:04:05"), incomingMaterial)
			}
		}
	}
}

func (asc *Ascend) pushTaskRun() {
	for {
		select {
		case incomingMaterial := <-asc.ProcessChan:
			fmt.Printf("[%s] incoming material received on processing chan: %v\n", time.Now().Format("2006-01-02 15:04:05"), incomingMaterial)
			for TokenPipe.TokenValid == false {
				time.Sleep(time.Millisecond * 10)
			}
			go asc.PostMessage(incomingMaterial)
			fmt.Printf("[%s] incoming material under process: %v\n", time.Now().Format("2006-01-02 15:04:05"), incomingMaterial)
		}
	}
}

func (asc *Ascend) PostMessage(pm *PostMessage) {
	reqData := asc.pipe.PrepareData(pm.JsonParam)
	pipeRsp := asc.pipe.RequestAction(pm.ApiUrl, reqData, TokenPipe.QueryTokenRsp.AccessToken)
	select {
	case pm.OutputChan <- pipeRsp:
		fmt.Println("response pass to client channel. ")
	default:
		fmt.Println("pass to client channel failed. ")
	}
}
